package com.wang.helloworld.storm.bolt;

import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class CommandMessageBolt extends BaseRichBolt{
	private static final long serialVersionUID = 4342418138794787741L;
	Logger logger = LoggerFactory.getLogger(getClass());
	
	OutputCollector collector;
	Map map;
	TopologyContext context;

	public void execute(Tuple tuple) {
		System.out.println("Receive command: " + tuple.getString(0) );

		// TODO: Filter command
		
		// Process chan
		collector.emit(tuple, new Values(tuple.getString(0)));

        collector.ack(tuple);
	}

	public void prepare(Map map, TopologyContext context, OutputCollector collector) {
		this.collector = collector;
		this.map = map;
		this.context = context;
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("instruction"));
	}
}
